/*
 * v1.1 by jetsum tonydon
 * netbeansIDE 7.1
 */
package txdnet.util.concurrent;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.*;

/**
 *
 * @author txdnet
 */
public class ReuseTaskExecutor implements ITaskExecutor{
    
    protected Queue taskQueue = new LinkedList();
    
    protected ThreadPoolExecutor threadPool = null;
    
    private ReturnQueueWaitHandler rejectedhandler = new ReturnQueueWaitHandler();
    
    protected int intBatchAddTaskNum = 32 ;
    
    private int periodms = 1000;
    
    public ReuseTaskExecutor(){
        rejectedhandler.setQueue(taskQueue);
    }
    
    public ReuseTaskExecutor(Queue taskQueue){
        this.taskQueue = taskQueue;
        this.rejectedhandler = new ReturnQueueWaitHandler();
        this.rejectedhandler.setQueue(this.taskQueue);
    }

    @Override
    public RejectedExecutionHandler getRejectedHandler() {
        return this.rejectedhandler;
    }
    
    int k=0;
    // 访问消息缓存的调度线程,将待处理任务批量执行
    final Runnable waitTaskThread = new Runnable(){
    @Override
    public void run()
    { 
        if(threadPool==null){return;}
    // 查看是否有待定请求，如果有，则批量提交 添加到线程池中
        while(k<intBatchAddTaskNum){
            if(!taskQueue.isEmpty()){
                   threadPool.execute((Runnable)getTaskQueue().poll());
            }
            k++;
        }
      k=0;
    }
    };

    // 调度线程池
    final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    final ScheduledFuture futureHandler = 
            scheduler.scheduleAtFixedRate(waitTaskThread, 0, periodms, TimeUnit.MILLISECONDS);
    
    @Override
    public void setThreadPoolExecutor(ThreadPoolExecutor threadPool) {
        this.threadPool = threadPool;
    }

    /**
     * @return the intBatchAddTaskNum
     */
    public int getIntBatchAddTaskNum() {
        return intBatchAddTaskNum;
    }

    /**
     * @param intBatchAddTaskNum the intBatchAddTaskNum to set
     */
    public void setIntBatchAddTaskNum(int intBatchAddTaskNum) {
        this.intBatchAddTaskNum = intBatchAddTaskNum;
    }

    /**
     * @return the periodms
     */
    public int getPeriodms() {
        return periodms;
    }

    /**
     * @param periodms the periodms to set
     */
    public void setPeriodms(int periodms) {
        this.periodms = periodms;
    }
    
    public int currQueueNum(){
        return this.getTaskQueue().size();
    }
    
    /*
     * 
     */
    @Override
    public void cancelAllTask(){
        if(scheduler!=null){
            futureHandler.cancel(true);
            scheduler.shutdown();
            getTaskQueue().clear();
        }
    }

    /**
     * @return the taskQueue
     */
    public Queue getTaskQueue() {
        return taskQueue;
    }

    /**
     * @param taskQueue the taskQueue to set
     */
    public void setTaskQueue(Queue taskQueue) {
        this.taskQueue = taskQueue;
    }

}